package slh.myMapReduce;

import net.minidev.json.JSONArray;
import net.minidev.json.JSONObject;
import net.minidev.json.JSONValue;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.BufferedReader;
import java.io.FileWriter;
import java.io.InputStreamReader;

public class ExportTopMoviesToJson {
    public static void main(String[] args) {
        Configuration conf = new Configuration();

        try {
            // 设置 HDFS 地址
            conf.set("fs.defaultFS", "hdfs://hadoop101:9000");
            FileSystem fs = FileSystem.get(conf);

            // 假设 MapReduce 输出路径
            Path outputPath = new Path("/user/niit/output/favouriteTop3");

            // 验证路径是否存在
            if (!fs.exists(outputPath)) {
                System.err.println("错误：输出路径不存在 - " + outputPath);
                System.exit(1);
            }

            // 用于保存前 3 名电影的集合
            JSONArray topMovies = new JSONArray();

            // 获取目录下所有文件
            FileStatus[] fileStatuses = fs.listStatus(outputPath);

            // 遍历结果文件
            for (FileStatus status : fileStatuses) {
                // 跳过目录和隐藏文件
                if (!status.isFile() || status.getPath().getName().startsWith("_")) {
                    continue;
                }

                // 读取文件内容
                try (BufferedReader reader = new BufferedReader(
                        new InputStreamReader(fs.open(status.getPath())))) {
                    String line;
                    while ((line = reader.readLine()) != null && topMovies.size() < 3) {
                        // 每行格式：电影 ID\t收藏量
                        String[] parts = line.split("\t");
                        if (parts.length < 2) continue;

                        // 创建电影 JSON 对象
                        JSONObject movie = new JSONObject();

                        movie.put("电影ID",parts[0]); // 示例产品名，可根据实际改
                        movie.put("收藏量", Integer.parseInt(parts[1]));
                        topMovies.add(movie);
                    }
                }
            }

            // 写入本地 JSON 文件（严格按积木报表示例格式：最外层 JSON 数组，内部对象格式匹配）
            try (FileWriter file = new FileWriter("top_movies.json")) {
                // 直接用 JSONValue 把 JSONArray 序列化为字符串写入，自动符合格式
                String jsonStr = JSONValue.toJSONString(topMovies);
                file.write(jsonStr);

                System.out.println("JSON 文件已生成：top_movies.json");
            }

        } catch (Exception e) {
            System.err.println("程序执行出错：" + e.getMessage());
            e.printStackTrace();
            System.exit(1);
        }
    }
}